package com.wfy.orchestra.dispatch

import android.util.Log
import com.wfy.orchestra.monitor.DefaultPerfMonitor
import com.wfy.orchestra.monitor.PerformanceMonitor
import com.wfy.orchestra.thread.ITaskExecutor
import com.wfy.orchestra.thread.PoolTaskExecutor
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.MainScope
import kotlinx.coroutines.channels.actor
import kotlinx.coroutines.launch
import java.util.LinkedList
import java.util.PriorityQueue
import java.util.Queue
import java.util.concurrent.ConcurrentHashMap
import kotlin.concurrent.Volatile

/**
 * @author : wangfengyuan
 * e-mail  : 446950883@qq.com
 * time    : 2025/3/21
 * desc    : Kotlin协程中的Channel和Actor模型适合处理事件驱动
 * 当有任务加入时发送事件，而不是轮询。这样当没有任务时，协程挂起，不会占用CPU。
 * version : 1.0
 */
open class EventTaskGraph(
    private val defaultExecutor: ITaskExecutor = PoolTaskExecutor(),
    private val mainScope: CoroutineScope = MainScope()
) : ITaskGraph {
    private val TAG = "TaskGraph"

    //单线程池，用于持续处理任务队列
    private val schedulerExecutor = MainScope()

    //线程安全的哈希表，存储所有任务节点（Key: 任务ID，Value: TaskNode）
    private val taskNodes: MutableMap<String, TaskNode> = ConcurrentHashMap()

    // 任务状态追踪
    private val runningTasks = ConcurrentHashMap<TaskNode, Boolean>()

    // 定义任务调度事件类型
    sealed class ScheduleEvent {
        data class AddTask(val node: TaskNode) : ScheduleEvent()
        data class TaskCompleted(val node: TaskNode) : ScheduleEvent()
        data class UpdatePriority(val taskId: String, val newPriority: Int) : ScheduleEvent()
    }

    // 使用 Actor 封装调度逻辑 Actor 按消息接收顺序处理，天然保证操作的原子性
    private val schedulerActor = mainScope.actor<ScheduleEvent>(Dispatchers.Main) {
        //优先级比较器（数值大的优先级高）
        val taskComparator = Comparator<TaskNode> { node1, node2 ->
            node2.getPriority().compareTo(node1.getPriority())
        }
        val priorityQueue = PriorityQueue(11, taskComparator) //优先级队列（存储待执行任务）,封装在Actor内部，确保串行访问
        for (event in channel) {
            when (event) {
                is ScheduleEvent.AddTask -> handleAddTask(event.node, priorityQueue)
                is ScheduleEvent.TaskCompleted -> handleTaskCompleted(event.node, priorityQueue)
                is ScheduleEvent.UpdatePriority -> handleUpdatePriority(event.taskId, event.newPriority, priorityQueue)
            }
        }
    }

    private fun handleAddTask(node: TaskNode, queue: PriorityQueue<TaskNode>) {
        queue.add(node)
        processNextTask(queue) // 尝试立即处理新任务
    }

    private fun handleTaskCompleted(node: TaskNode, queue: PriorityQueue<TaskNode>) {
        node.getSuccessors().forEach { successor ->
            // 减少后继任务的入度，若入度为0则加入队列
            if (taskNodes[successor]!!.decrementInDegree() == 0) {
                queue.add(taskNodes[successor])
            }
        }
        processNextTask(queue) // 处理后续任务
    }

    private fun handleUpdatePriority(
        taskId: String, newPriority: Int, queue: PriorityQueue<TaskNode>
    ) {
        taskNodes[taskId]?.let { node ->
            queue.remove(node)
            node.task.priority = newPriority // 需修改Task为可变优先级
            queue.add(node)
            processNextTask(queue)
        }
    }

    private fun processNextTask(queue: PriorityQueue<TaskNode>) {
        val node = queue.poll()
        if (node != null) {
            submitTask(node)
        }
    }

    @Volatile
    private var isCancelled = false //任务取消标志（volatile 保证多线程可见性）
    private var perfMonitor: PerformanceMonitor = DefaultPerfMonitor() //性能监控器
    private var errorListener: OnErrorListener? = null //错误监听器

    /**
     * 添加任务到图中
     */
    override fun addTask(task: ITask) {
        val v = taskNodes[task.id]
        if (v == null) {
            taskNodes[task.id] = TaskNode(task)
        }
    }

    override fun addTask(vararg tasks: ITask) {
        tasks.forEach { addTask(it) }
    }

    /**
     * 添加任务依赖关系
     */
    override fun addDependency(taskId: String, dependencyId: String) {
        val taskNode = taskNodes[taskId] ?: throw IllegalArgumentException("ITask with ID $taskId does not exist")
        val depNode = taskNodes[dependencyId] ?: throw IllegalArgumentException("Dependency with ID $dependencyId does not exist")
        depNode.addSuccessor(taskId)// 将taskId任务添加为dependencyId任务的后继
        taskNode.incrementInDegree()// 增加当前任务的入度
    }

    override fun addDependency(taskId: String, dependencyIds: List<String>) {
        dependencyIds.forEach { addDependency(taskId, it) }
    }

    override fun configureDependencies(block: TaskDslBuilder.() -> Unit) {
        TaskDslBuilder(this).apply(block)
    }

    override fun start() {
        Log.v(TAG, "taskNodes $taskNodes")//循环依赖时StackOverflowError,successors改为string可以解决
        check(!hasCycle()) { "Cycle detected in task graph" } // 检查循环依赖
        //启动持续调度
        schedulerExecutor.launch {
            taskNodes.values.filter { it.getInDegree() == 0 }.forEach {
                schedulerActor.send(ScheduleEvent.AddTask(it))
            }
        }
    }

    private fun submitTask(node: TaskNode) {
        if (isCancelled) {// 检查全局取消状态
            mainScope.launch { node.task.onCancelled() }
            return
        }
        runningTasks[node] = true
        val executor: ITaskExecutor = if (node.task.getExecutor() != null) node.task.getExecutor()!! else defaultExecutor
        perfMonitor.onTaskSubmit(node.task)
        executor.execute {
            //性能监控
            perfMonitor.onTaskStarted(node.task)
            try {
                // 双重检查
                if (!isCancelled) {
                    node.task.execute()
                }
            } catch (e: Exception) {
                //错误处理机制
                Log.e(TAG, "${node.task} execute error ${e.message}")
                if (errorListener != null) {
                    mainScope.launch {
                        errorListener!!.onError(node.task, e)
//                        cancelDependentTasks(node) //取消依赖此任务的后继
                    }
                }
                isCancelled = true
            } finally {
                perfMonitor.onTaskFinished(node.task)
                runningTasks.remove(node)
                // 确保生命周期销毁后不回调主线程
                if (!isCancelled) {
                    mainScope.launch { onTaskCompleted(node) }
                }
            }
        }
    }

    // 公开的取消方法
    override fun cancelAll() {
        isCancelled = true
        defaultExecutor.shutdown()
        schedulerActor.close()// 关闭 Actor 通道
        taskNodes.values.forEach { it.task.onCancelled() }// 取消所有任务
        taskNodes.clear()// 清空任务节点
    }

    private fun onTaskCompleted(node: TaskNode) {
        MainScope().launch {
            schedulerActor.send(ScheduleEvent.TaskCompleted(node))
        }
    }

    /**
     * 循环依赖检测
     */
    private fun hasCycle(): Boolean {
        // 使用普通队列即可（拓扑排序不依赖优先级）
        val queue: Queue<TaskNode> = LinkedList()
        // 入度表
        val inDegreeMap: MutableMap<TaskNode, Int> = HashMap()

        // 初始化入度表
        for (node in taskNodes.values) {
            inDegreeMap[node] = node.getInDegree()
            if (inDegreeMap[node] == 0) {
                queue.add(node)
            }
        }

        var processedCount = 0
        while (!queue.isEmpty()) {
            val node = queue.poll()
            processedCount++
            // 减少所有后继的入度
            for (successor in node.getSuccessors()) {
                val successorTask = taskNodes[successor]
                val newInDegree = inDegreeMap[successorTask]!! - 1
                inDegreeMap[successorTask!!] = newInDegree
                if (newInDegree == 0) {
                    queue.add(successorTask)
                }
            }
        }
        // 如果处理的任务数 ≠ 总任务数 → 存在环
        return processedCount != taskNodes.size
    }

    //依赖可视化,生成DAG的Graphviz描述
    override fun generateDotGraph(): String {
        val sb = StringBuilder("digraph G {\n")
        taskNodes.values.forEach { node ->
            node.getSuccessors().forEach {
                sb.append("  \"${node.task.id}\" -> \"${it}\";\n")
            }
        }
        return sb.append("}").toString()
    }

    override fun updatePriority(taskId: String, newPriority: Int) {
        MainScope().launch {
            schedulerActor.send(ScheduleEvent.UpdatePriority(taskId, newPriority))
        }
    }

    override fun setOnErrorListener(listener: OnErrorListener) {
        this.errorListener = listener
    }

    override fun setPerformanceMonitor(monitor: PerformanceMonitor) {
        this.perfMonitor = monitor
    }
}